package cn.hesion.Streaming.kafka

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

import java.util.Properties

/**
 * ClassName: KafkaProducer <br/>
 * Description: <br/>
 * date: 2021/7/22 9:46<br/>
 *
 * @author Hesion<br/>
 * @version
 * @since JDK 1.8
 */
object KafkaProducer {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)

    val lines = sc.textFile("data/sample.log")
    val prop = new Properties()
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux121:9092")
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])

    lines.foreachPartition { iter =>
      //初始化 kafkaProducer
      val producer = new KafkaProducer[String, String](prop)
      iter.foreach { line =>
        // 封装数据
        val record = new ProducerRecord[String, String]("mytopic1", line)
        // 发送数据
        producer.send(record)
      }
      producer.close()
    }
  }
}
